package com.aotain.nyx.vc;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.nyx.flink.BasicTuple;


/**
 * 
     *@Title:暴力破解MAP
     *@Description:
     *@Author:Turk
     *@Since:2017年7月13日
     *@Version:1.1.0
 */
public class VCMap implements MapFunction<Tuple2<String,String>,Tuple2<String,BasicTuple>>{

	/**
	  *
	  */
	    
	private static final long serialVersionUID = 2400652295112006863L;

	@Override
	public Tuple2<String, BasicTuple> map(Tuple2<String, String> value)
			throws Exception {
		
//		// [0-4]dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64+ "|" + 
		// [5-9]domainname+ "|" +url+ "|" +type+ "|" +delaytime+ "|" +idcno+ "|" + 
		// [10-14]upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|" + 
		// [15-19]dnStreamOctets+ "|" +dnStreamPacket+ "|" + refer + "|" +post | OUT/IN
		
		Tuple2<String, BasicTuple> ret = new Tuple2<String, BasicTuple>();
		String logString = value.f1; //格式化后的原始数据
		String[] items = logString.split("\\|",-1);
		if(items.length <= 18)
			return ret;
		
		String ispost = items[18];
		if(ispost.equals("0"))
			return ret;
		
		String dip = items[0];
		String sip = items[2];
		String gis = items[4];
		String idc = items[9];
		String flowdirection = items[19];
		
		String tripleKey = dip + "|" + sip + "|" + idc;
		BasicTuple tuple = new BasicTuple();
		tuple.setDestIP(dip);
		tuple.setSourceIP(sip);
		tuple.setGIS(gis);
		
		long upStreamOctets = 0; //UPSTREAMOCTETS
		long upStreamPacket = 0; //UPSTREAMPACKET			
		upStreamOctets = Long.parseLong(items[10]);
		upStreamPacket = Long.parseLong(items[11]);
		tuple.setUpStreamOctets(upStreamOctets);
		tuple.setUpStreamPacket(upStreamPacket);
		tuple.setSessionNum(1l);
		tuple.setFlowDirection(flowdirection);
		
		ret.setFields(tripleKey, tuple);
		
		return ret;
		    
	}

}

    